package com.sharp.mq.service.delay;

import com.sharp.mq.bo.RabbitMqMessage;
import com.sharp.mq.config.DynamicConsumerContainerFactory;
import com.sharp.mq.config.MessageDelegate;
import com.sharp.mq.service.RabbitMqService;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;

/**
 * Title: BaseDelayMessageService
 * Description: Copyright: Copyright (c) 2019 Company: BHFAE
 *
 * @author Sharp
 * @date 2021/7/5 10:01
 */
public abstract class BaseDelayMessageService<T> {
    private static final String DELAY_QUEUE_NAME_PREFIX = "delay.queue.";
    private static final String DELAY_EXCHANGE_NAME_PREFIX = "delay.plugIn.exchange.";
    private static final String DELAY_ROUTING_KEY_PREFIX = "delay.routing.key.";

    @Resource
    private ConnectionFactory connectionFactory;
    @Resource
    private RabbitAdmin rabbitAdmin;
    @Resource
    private RabbitMqService rabbitMqService;

    private Queue queue;
    private Exchange exchange;
    private Binding binding;

    @PostConstruct
    public void init() {
        this.queue = buildQueue();
        this.exchange = buildDelayPlugInExchange();
        this.binding = delayPlugInMessageBinding(exchange, queue);
        DynamicConsumerContainerFactory.builder().connectionFactory(connectionFactory).rabbitAdmin(rabbitAdmin)
                .autoAck(false).queue(buildQueue()).exchange(buildDelayPlugInExchange()).binding(binding)
                .messageDelegate(new MessageDelegate<T>() {
                    @Override
                    protected void handleMessage(T messageBody) {
                        delayMessageHandler(messageBody);
                    }
                }).build().getObject().start();

    }

    private Queue buildQueue() {
        return new Queue(DELAY_QUEUE_NAME_PREFIX + this.getClass().getName(), false, false, false);
    }

    private CustomExchange buildDelayPlugInExchange() {
        //创建一个自定义交换机，可以发送延迟消息
        Map<String, Object> args = new HashMap<>(1);
        args.put("x-delayed-type", "direct");
        return new CustomExchange(DELAY_EXCHANGE_NAME_PREFIX + this.getClass().getName(), "x-delayed-message", false, false, args);
    }

    private Binding delayPlugInMessageBinding(Exchange delayPlugInExchange, Queue delayPlugInMessageQueue) {
        return BindingBuilder.bind(delayPlugInMessageQueue).to(delayPlugInExchange).with(DELAY_ROUTING_KEY_PREFIX + this.getClass().getName()).noargs();
    }

    /**
     * 延迟消息处理
     *
     * @param message 消息
     */
    protected abstract void delayMessageHandler(T message);

    public void sendDelayMessage(T message, long delay) {
        RabbitMqMessage rabbitMqMessage = new RabbitMqMessage();
        rabbitMqMessage.setQueueName(this.queue.getName());
        rabbitMqMessage.setExchangeName(this.exchange.getName());
        rabbitMqMessage.setRouterKey(this.binding.getRoutingKey());
        rabbitMqMessage.setBody(message);
        rabbitMqMessage.setDelay(delay);
        rabbitMqService.sendMessage(rabbitMqMessage);
    }
}
